package com.spark.ml.demo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.{SparkConf, SparkContext}
import org.jblas.DoubleMatrix

object ALS1{
  // 定义相似度函数，cosine相似
  def cosineSimilarity(vec1: DoubleMatrix, vec2: DoubleMatrix): Double = {
    vec1.dot(vec2) / (vec1.norm2() * vec2.norm2())//xy连续和/x连续平方 和 再开方 乘以 y连续平方 和 再开方
  }
  def loadModel(sc : SparkContext,path : String) : MatrixFactorizationModel ={
    MatrixFactorizationModel.load(sc,path)
  }
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("als1").setMaster("local")
    val sc = new SparkContext(conf)
    Logger.getRootLogger.setLevel(Level.WARN)
    //load model
    //val sameModel = loadModel(sc,"")
    //read sample data

    val data = sc.textFile(this.getClass.getClassLoader.getResource("alsdata.txt").getFile)//Users/qifei/work/alsdata.txt
    val ratings = data.map(_.split(",")match {
      case Array(user,item,rate) =>
        Rating(user.toInt,item.toInt,rate.toDouble)
    })

    //create model
    val rank = 10
    val numIterations = 10
    val model = ALS.train(ratings,rank,numIterations,0.01)

    //predict result
    val usersProducts=ratings.map{
      case Rating(user,product,rate) =>
        (user,product)
    }
    val predictions = model.predict(usersProducts).map{
      case Rating(user,product,rate) =>
        ((user,product),rate)
    }
    val ratesAndPreds = ratings.map{
      case Rating(user,product,rate) =>
        ((user,product),rate)
    }join(predictions)

    val MSE = ratesAndPreds.map{
      case ((user,product),(r1,r2)) =>
        val err = (r1-r2)
        err*err
    }.mean()

    //获取某个用户所有打分记录,比如用户1
    val productForUser = ratings.keyBy(_.user).lookup(1)
    println("用户1在模型中的原始打分记录")
    println(productForUser.mkString("\n"))

    //recommend product for old user，获取某个用户推荐topN 广告，以及其评分
    val p = model.recommendProducts(1,4)
    println("给用户1，推荐三个广告")
    println(p.mkString("\n"))
    //广告推荐
    // 选定id为2的广告
    val itemId = 2
    // 获取该物品的隐因子向量
    val itemFactor = model.productFeatures.lookup(itemId).head
    // 将该向量转换为jblas矩阵类型
    val itemVector = new DoubleMatrix(itemFactor)
    // 计算广告2 与其他广告的相似度
    val sims = model.productFeatures.map{ case (id, factor) =>
      val factorVector = new DoubleMatrix(factor)
      val sim = cosineSimilarity(factorVector, itemVector)
      (id, sim)
    }
    // 获取与广告2最相似的3个广告
    val sortedSims = sims.top(3)(Ordering.by[(Int, Double), Double] { case (id, similarity) => similarity })
    // 打印结果
    println("从推荐模型中读取与广告2最相似的三个广告")
    println(sortedSims.mkString("\n"))

    println("该模型误差平方和（Mean Squared Error）= "+MSE)
    println("用户ID" + "\t" + "广告ID" + "\t" + "真实打分" + "\t" + "模型预测值")
    ratesAndPreds.collect.foreach(
      rating => {
        println(rating._1._1 + "   \t" + rating._1._2 + " \t    " + rating._2._1 + "\t   " + rating._2._2)
      }
    )
    //save model
    val modelPath = "/Users/qifei/work/machine"
    //model.save(sc,modelPath)
  }
}

